package com.tencent.albert.weixin.driver;

import com.tencent.albert.weixin.driver.bean.AccessToken;
import com.tencent.albert.weixin.driver.bean.ReceivedMsgInfo;
import com.tencent.albert.weixin.driver.bean.SendMsgInfo;
import com.tencent.albert.weixin.driver.dao.AccessTokenDao;
import com.tencent.albert.weixin.driver.dao.MsgDao;
import com.tencent.albert.weixin.driver.utils.WeixinUtil;
import org.apache.tomcat.jdbc.pool.DataSource;
import org.apache.tomcat.jdbc.pool.PoolProperties;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.sql.Connection;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Future;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.TimeUnit;

/**
 * Created by albertzhu on 15/2/13.
 */
public class MsgManager {

    static AccessTokenDao accessTokenDao;
    static MsgDao msgDao;

    SaveMsgWorker saveMsgWorker;
    LoadSendMsgWorker loadSendMsgWorker;
    HandlerSendedMsgWorker handlerSendedMsgWorker;
    SaveHandledMsgWorker saveHandledMsgWorker;

    public void init() {

        accessTokenDao = new AccessTokenDao();
        msgDao = new MsgDao();
        startRefreTokenThread();

        saveMsgWorker = new SaveMsgWorker();
        saveMsgWorker.setName("saveMsgWorker");
        saveMsgWorker.start();
        loadSendMsgWorker = new LoadSendMsgWorker();
        loadSendMsgWorker.setName("loadSendMsgWorker");
        loadSendMsgWorker.start();
        handlerSendedMsgWorker = new HandlerSendedMsgWorker();
        handlerSendedMsgWorker.setName("handlerSendedMsgWorker");
        handlerSendedMsgWorker.start();
        saveHandledMsgWorker = new SaveHandledMsgWorker();
        saveHandledMsgWorker.setName("saveHandledMsgWorker");
        saveHandledMsgWorker.start();

    }

    private static Logger logger = LoggerFactory.getLogger(MsgManager.class);

    public static AccessToken getAccessToken() {
        return accessToken;
    }

    public static void setAccessToken(AccessToken accessToken) {
        MsgManager.accessToken = accessToken;
    }

    private static AccessToken accessToken;

    private static void refreAccessToken() {
        AccessToken temp = accessTokenDao.getAccessToken(0);
        if (null != temp) {
            accessToken = temp;
        }
        temp = WeixinUtil.getAccessToken(Constants.appId, Constants.appSecret);
        if (null != temp) {
            if (null != accessToken ) {
                temp.setId(accessToken.getId());
            }
            accessToken = temp;
            accessTokenDao.updateAccessToken(accessToken);
        }
    }

    private static void startRefreTokenThread() {
        Thread t = new Thread() {
            @Override
            public void run() {
                while (true) {
                    msgDao.createTable();
                    refreAccessToken();
                    try {
                        Thread.sleep(5 * 60 * 1000);
                    } catch (InterruptedException ex) {
                        logger.info("ConfigManager", "refreAccessToken sleep error " + ex, ex);
                    }

                }
            }
        };
        t.setName("RefreTokenThread");
        t.start();
        logger.info("init refreTokenThread finished ");
    }

    private static DataSource datasource;

    public static Connection getConn() {
        Future<Connection> future = null;
        try {
            future = datasource.getConnectionAsync();
            int i = 0;
            while (!future.isDone()) {
                if (i > 4 ) {
                    return null;
                }
                try {
                    Thread.sleep(100);
                }catch (InterruptedException x) {
                    Thread.currentThread().interrupt();
                }
                i++;
            }
            Connection con = future.get(); //should return instantly
            return con;
        } catch (Exception e) {
            logger.info("get DBConn failed "+e,e);
        }
        return null;

    }


    static LinkedBlockingDeque<ReceivedMsgInfo> receivedMsgInfos = new LinkedBlockingDeque<ReceivedMsgInfo>();

    static LinkedBlockingDeque<SendMsgInfo> sendMsgInfos = new LinkedBlockingDeque<SendMsgInfo>();

    static LinkedBlockingDeque<SendMsgInfo> handledMsgInfos = new LinkedBlockingDeque<SendMsgInfo>();

    public static void addReceivedMsg(ReceivedMsgInfo msgInfo) {
        receivedMsgInfos.add(msgInfo);
    }

    public void initDbConnPool(String jdbdUrl, String jdbcUser, String jdbcPass, int initialPoolSize, int maxActive) {
        PoolProperties p = new PoolProperties();
        p.setUrl(jdbdUrl);
        p.setDriverClassName("com.mysql.jdbc.Driver");
        p.setUsername(jdbcUser);
        p.setPassword(jdbcPass);
        p.setJmxEnabled(true);
        p.setTestWhileIdle(false);
        p.setTestOnBorrow(true);
        p.setValidationQuery("SELECT 1");
        p.setTestOnReturn(false);
        p.setValidationInterval(30000);
        p.setTimeBetweenEvictionRunsMillis(30000);
        p.setMaxActive(maxActive);
        p.setInitialSize(initialPoolSize);
        p.setMaxWait(10000);
        p.setRemoveAbandonedTimeout(60);
        p.setMinEvictableIdleTimeMillis(30000);
        p.setMinIdle(10);
        p.setLogAbandoned(true);
        p.setRemoveAbandoned(true);
        p.setJdbcInterceptors("org.apache.tomcat.jdbc.pool.interceptor.ConnectionState;"+
                "org.apache.tomcat.jdbc.pool.interceptor.StatementFinalizer");
        datasource = new DataSource();
        datasource.setPoolProperties(p);
        logger.info("init dbPool finished ");
    }

    class SaveMsgWorker extends Thread {
        public void run() {
            List<ReceivedMsgInfo> list = new ArrayList<ReceivedMsgInfo>();
            while (true) {
                try {
                    ReceivedMsgInfo info = receivedMsgInfos.take();
                    if (null != info) {
                        list.add(info);
                        logger.debug("add "+info.getResponseContent());
                    }
                    for (int i = 0 ; i< 20 ; i++) {
                        info = receivedMsgInfos.poll(10, TimeUnit.MILLISECONDS);
                        if (null != info) {
                            list.add(info);
                            logger.debug("add "+info.getResponseContent());
                        }
                    }
                    saveReceivedMsg(list);
                    list.clear();
                } catch (Throwable e) {
                    logger.info("run save msg thread error " + e, e);
                }
            }

        }
    }

    class LoadSendMsgWorker extends Thread {
        public void run() {
            while(true) {
                int size = loadSendedMsg();
                try {
                    if (size == 0 ) {
                        Thread.sleep(1000);
                    } else if (size < 10) {
                        Thread.sleep(100);
                    } else {
                        Thread.sleep(10);
                    }
                }catch (InterruptedException x) {
                    Thread.currentThread().interrupt();
                }
            }
        }
    }

    class HandlerSendedMsgWorker extends Thread {
        public void run() {
            while(true) {
                try {
                    SendMsgInfo info = sendMsgInfos.take();
                    if (null != info) {
                        WeixinUtil.sendRespMsg(info);
                        handledMsgInfos.add(info);
                    }
                } catch (Throwable e) {
                    logger.info("run handler msg thread error " + e, e);
                }
            }
        }
    }

    class SaveHandledMsgWorker extends Thread {
        public void run() {
            List<SendMsgInfo> list = new ArrayList<SendMsgInfo>();
            while (true) {
                try {
                    SendMsgInfo info = handledMsgInfos.take();
                    if (null != info) {
                        list.add(info);
                    }
                    for (int i = 0 ; i< 20 ; i++) {
                        info = handledMsgInfos.poll(10, TimeUnit.MILLISECONDS);
                        if (null != info) {
                            list.add(info);
                        }
                    }
                    updateSendMsg(list);
                    list.clear();
                } catch (Throwable e) {
                    logger.info("run update sened msg thread error " + e, e);
                }
            }

        }
    }

    /**
     * 批量把接收到的数据写入数据库
     * @param infos
     */
    static void saveReceivedMsg(List<ReceivedMsgInfo> infos) {
        msgDao.insertMsgLog(infos);
    }

    /**
     * 加载待发数据
     */
    static int loadSendedMsg() {
        List<SendMsgInfo> infos = msgDao.getSendMsgInfos(100);
        if (infos.size() > 0) {
            sendMsgInfos.addAll(infos);
        }
        return infos.size();
    }

    /**
     * 将待发数据状态更新回数据库
     * @param infos
     */
    static void updateSendMsg(List<SendMsgInfo> infos) {
        msgDao.updateSendedMsg(infos);
    }

}
